package com.example.flinkcourse.lesson1.sink;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.flink.streaming.connectors.elasticsearch7.ElasticsearchSink;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Elasticsearch Sink 工厂类
 * 用于创建 Elasticsearch 输出
 */
public class EsSinkFactory {
    private static final Logger LOG = LoggerFactory.getLogger(EsSinkFactory.class);

    /**
     * 创建 Elasticsearch Sink
     * @return ElasticsearchSink<String>
     */
    public static ElasticsearchSink<String> createEsSink() {
        LOG.info("Creating Elasticsearch sink...");

        // 配置 Elasticsearch 主机
        List<HttpHost> httpHosts = new ArrayList<>();
        httpHosts.add(new HttpHost("localhost", 9200, "http"));

        // 配置 Elasticsearch Sink
        ElasticsearchSink.Builder<String> esSinkBuilder = new ElasticsearchSink.Builder<>(
            httpHosts,
            new ElasticsearchSinkFunction<String>() {
                @Override
                public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
                    // 创建索引请求
                    Map<String, String> json = new HashMap<>();
                    json.put("data", element);
                    json.put("timestamp", String.valueOf(System.currentTimeMillis()));

                    IndexRequest indexRequest = Requests.indexRequest()
                        .index("flink-output")
                        .source(json);

                    indexer.add(indexRequest);
                }
            }
        );

        // 配置批量请求
        esSinkBuilder.setBulkFlushMaxActions(1000);
        esSinkBuilder.setBulkFlushMaxSizeMb(5);
        esSinkBuilder.setBulkFlushInterval(5000);

        // 如果启用了安全认证
        if (isSecurityEnabled()) {
            configureSecurity(esSinkBuilder);
        }

        LOG.info("Elasticsearch sink created successfully");
        return esSinkBuilder.build();
    }

    /**
     * 检查是否启用安全认证
     */
    private static boolean isSecurityEnabled() {
        // 从配置文件中读取是否启用安全认证
        return false; // 默认不启用
    }

    /**
     * 配置安全认证
     */
    private static void configureSecurity(ElasticsearchSink.Builder<String> builder) {
        // 配置基本认证
        Map<String, String> config = new HashMap<>();
        config.put("elasticsearch.security.authentication", "basic");
        config.put("elasticsearch.security.username", "elastic");
        config.put("elasticsearch.security.password", "changeme");
        builder.setRestClientFactory(new RestClientFactory(config));
    }
} 